using System;
using System.Linq;
using System.Collections.Generic;
using Apache.Cassandra;
using Cassandraemon.Builder;
using Cassandraemon.FullText;


namespace Cassandraemon.Operators
{


	class FullTextSearchCassandraOperator<T> : CassandraOperatorBase<T>
	{
		public IAnalyzer Analyzer { get; set; }
		
		public List<string> Matches { get; set; }
		
		public List<string> UnMatches { get; set; }
		
		public string ColumnFamily { get; set; }
		
		public ColumnParent ColumnParent { get; set; }
		
		public List<byte[]> Fields { get; set; }
		
		public SlicePredicate  Predicate { get; set; }
		
		public ConsistencyLevel ConsistencyLevel { get; set; }
	
	
		public FullTextSearchCassandraOperator () { }
		
		
		protected override object OperateInner ()
		{
			var fields = new List<string>(Fields.ConvertAll(x => x.ToUTF8()));
			
			var matchDic = Search(Matches, fields, false);
			
			if(matchDic.Count == 0) 
				return new List<CassandraEntity<List<SuperColumn>>>();
			
			var unmatchDic = Search(UnMatches, matchDic.Keys.ToList(), true);
			
			foreach(var key in unmatchDic.Keys)
			{
				if(matchDic.ContainsKey(key))
				{
					var unmatchColumns = unmatchDic[key];
					foreach(var c in unmatchColumns)
					{
						matchDic[key].RemoveAll(x => x.Name.SequenceEqual(c.Name));
					}
					if(matchDic[key].Count == 0) matchDic.Remove(key);
				}
			}
			
			
			var superColumnList = new List<SuperColumn>();
			foreach(var key in matchDic.Keys)
			{
				superColumnList.Add(key, matchDic[key]);
			}
			
			var entity = new CassandraEntity<List<SuperColumn>>();
			entity.Context = this.Context;
			entity.ColumnFamily = this.ColumnParent.Column_family;
			entity.Data = superColumnList;
		
			return new List<CassandraEntity<List<SuperColumn>>> { entity };
		}

		private Dictionary<string, List<Column>> Search(List<string> matches, List<string> fields, bool forNegativeSearch)
		{
			var dic = new Dictionary<string, List<Column>>();
			
			foreach(var match in matches)
			{
				var terms = Analyzer.SplitQueryPhrase(match);
				
				var fieldKeyDic = new Dictionary<string, List<Column>>();
				
				int offset = 0;
				
				foreach(string term in terms)
				{
					foreach(string field in fields)
					{
						var parent = new ColumnParent();
						parent.Column_family = this.ColumnFamily;
						parent.Super_column = field.ToCassandraByte();
						
						var predicate = new SlicePredicate();
						if(fieldKeyDic.ContainsKey(field) && fieldKeyDic[field].Count != 0)
						{
							predicate.Column_names = fieldKeyDic[field].ConvertAll(x => x.Name);
						}
						else
						{
							predicate.Slice_range = new SliceRange();
							predicate.Slice_range.Start = new byte[]{};
							predicate.Slice_range.Finish = new byte[]{};
							predicate.Slice_range.Reversed = false;
							predicate.Slice_range.Count = int.MaxValue;
						}
						
						var list = Client.get_slice(term.ToCassandraByte(),
									                 parent,
									                 predicate,
									                 this.ConsistencyLevel);
									                 
						var columnList = new List<Column>(list.ConvertAll(x => x.Column));
						
						if(!fieldKeyDic.ContainsKey(field)) 
						{
							fieldKeyDic[field] = new List<Column>();
						}
						
						var columns = fieldKeyDic[field];
						
						if(columns.Count() == 0)
						{
							columns.AddRange(columnList);
						}
						else
						{
							var deleteColumn = new List<Column>();
							foreach(var c in columns)
							{
								var tmpColumn = columnList.FirstOrDefault(x => x.Name.SequenceEqual(c.Name));
								if(tmpColumn == default(Column))
								{
									deleteColumn.Add(c);
									continue;
								}
								List<int> l = c.Value.ToObject<List<int>>();
								List<int> tmpList = tmpColumn.Value.ToObject<List<int>>();
								List<int> newList = new List<int>();
								
								foreach(int index in l)
								{
									if(tmpList.Contains(index + offset))
									{
										newList.Add(index);
									}
								}
								
								c.Value = newList.ToCassandraByte();
							}
							deleteColumn.ForEach(x => columns.Remove(x));
						}
						
					}
					
					offset++;

					if(!forNegativeSearch)
					{
						if(fields.All(x => !fieldKeyDic.ContainsKey(x) || 
						              		fieldKeyDic[x].Count == 0))
						{
							return new Dictionary<string, List<Column>>();
						}
					}
				}
				
				foreach(var key in fieldKeyDic.Keys)
				{
					if(!dic.ContainsKey(key))
					{
						dic[key] = fieldKeyDic[key];
					}
					else
					{
						var l = dic[key];
						var tempList = fieldKeyDic[key];
						var newList = new List<Column>();
						
						if(!forNegativeSearch)
						{
							foreach(Column c in l)
							{
								if(tempList.FindIndex(x => x.Name.SequenceEqual(c.Name)) != -1)
								{
									newList.Add(c);
								}
							}
						}
						else
						{
							newList.AddRange(l);
							tempList.ForEach( x => 
		                 	{
			                	if(newList.FindIndex( y => y.Name.SequenceEqual(x.Name) ) == -1)
								{
									newList.Add(x);
								}
							});
						}
						
						if(newList.Count == 0)
						{
							dic.Remove(key);
							fields.Remove(key);
						}
						else
						{
							dic[key] = newList;
						}
					}
				}
			}
			
			var deleteField = new List<string>();
			foreach(var key in dic.Keys)
			{
				if(dic[key].Count == 0) deleteField.Add(key);
			}
			deleteField.ForEach(x => dic.Remove(x));
			
			return dic;
		}

		public static ICassandraOperator<T> Create(CassandraOperatorCondition<T> condition)
		{
			if(condition.IsSetKey) throw new InvalidOperationException("You can't specify Key in case of fulltext_search operation.");
			if(condition.IsSetKeys) throw new InvalidOperationException("You can't specify Keys in case of fulltext_search operation.");
			
			if(condition.IsSetStartKey || 
			   condition.IsSetFinishKey || 
			   condition.IsSetStartToken || 
			   condition.IsSetFinishToken ||
			   condition.IsSetKeyCount)
				throw new InvalidOperationException("You can't specify range key in case of fulltext_search operation.");
	
			if(condition.IsSetColumns &&
			  (condition.IsSetStartColumn || 
			   condition.IsSetFinishColumn || 
			   condition.IsSetReverse ||
			   condition.IsSetCount))
				throw new InvalidOperationException("You can't specify both Columns and range column in case of fulltext_search operation.");
			
			
			var oper = new FullTextSearchCassandraOperator<T>();
			
			oper.Analyzer = condition.Analyzer;
			oper.Matches = condition.Matches;
			oper.UnMatches = condition.UnMatches;
			oper.ColumnFamily = condition.ColumnFamily;
			oper.Fields = condition.Columns;
			oper.ConsistencyLevel = condition.ConsistencyLevel;
			oper.PostProcessor = condition.PostProcessor;
			
			oper.ColumnParent = new ColumnParent();
			oper.ColumnParent.Column_family = condition.ColumnFamily;
			
			oper.Predicate = new SlicePredicate();
			oper.Predicate.Column_names = condition.Columns;
			if(condition.Columns == null)
			{
				oper.Predicate.Slice_range = new SliceRange();
				oper.Predicate.Slice_range.Start = condition.StartColumn;
				oper.Predicate.Slice_range.Finish = condition.FinishColumn;
				oper.Predicate.Slice_range.Reversed = condition.Reverse;
				oper.Predicate.Slice_range.Count = condition.Count;
			}
			
			
			Type type = typeof(T);
			
			if(type.Equals(typeof(List<SuperColumn>)))
			{
				if(condition.IsSetSuperColumn) throw new InvalidOperationException("You can't specify SuperColumn in case of fulltext_search SuperColumn.");
			}
			else
			{
				throw new InvalidOperationException("You must specify List<SuperColumn> to CassandraEntity's GenericType in fulltext_search operation."); 
			}
			
			return oper;
		}

	}
}
